package com.atguigu.edu.app.dwd.db;

import com.atguigu.edu.app.BaseSQLApp;
import com.atguigu.edu.common.Constant;
import com.atguigu.edu.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class Dwd_03_TradeOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_03_TradeOrderDetail().init(
                3003,2,
                "Dwd_03_TradeOrderDetail"
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db 数据
        readOdsDb(tEnv, "Dwd_03_TradeOrderDetail");

        // 2. 过滤出所需数据
        filterOdsData(tEnv);

        // 3.表join
        join(tEnv);

        // 4.写出到kafka
        writeToKafka(tEnv);
    }

    private void writeToKafka(StreamTableEnvironment tEnv) {
        tEnv.executeSql("create table dwd_trade_order_detail(" +
                "id string," +
                "order_id string," +
                "user_id string," +
                "course_id string," +
                "course_name string," +
                "province_id string," +
                "date_id string," +
                "create_time string," +
                "source_id string," +
                "origin_amount string," +
                "coupon_reduce string," +
                "final_amount string," +
                "ts bigint, " +
                "primary key(id) not enforced)" + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
        tEnv.executeSql("insert into dwd_trade_order_detail select * from joinResult");
    }

    private void join(StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        Table joinResult = tEnv.sqlQuery("select " +
                "od.id," +
                "od.order_id," +
                "od.user_id," +
                "od.course_id," +
                "od.course_name," +
                "oi.province_id," +
                "od.date_id," +
                "od.create_time," +
                "si.source_id," +
                "od.origin_amount," +
                "od.coupon_reduce," +
                "od.final_amount," +
                "od.ts " +
                "from order_detail od " +
                "join order_info oi on od.order_id=oi.id " +
                "left join source_info si on oi.session_id=si.session_id ");

//        joinResult.execute().print();
        tEnv.createTemporaryView("joinResult",joinResult);
    }

    private void filterOdsData(StreamTableEnvironment tEnv) {
        // order_detail
        Table orderDetail = tEnv.sqlQuery("select " +
                "data['id'] id," +
                "data['order_id'] order_id," +
                "data['user_id'] user_id," +
                "data['course_id'] course_id," +
                "data['course_name'] course_name," +
                "data['create_time'] create_time," +
                "   date_format(`data`['create_time'],'yyyy-MM-dd') date_id," +
                "data['origin_amount'] origin_amount," +
                "data['coupon_reduce'] coupon_reduce," +
                "data['final_amount'] final_amount," +
                "ts," +
                "pt " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_detail' " +
                "and `type`='insert' ");

//        orderDetail.execute().print();
        tEnv.createTemporaryView("order_detail", orderDetail);

        // order_info
        Table orderInfo = tEnv.sqlQuery("select " +
                "data['id'] id," +
                "data['province_id'] province_id ," +
                "data['session_id'] session_id " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_info' " +
                "and `type`='insert' ");

//        orderInfo.execute().print();
        tEnv.createTemporaryView("order_info", orderInfo);

        // source_info(通过 session_id 在 log 中找到 source_id)
        readOdsLog(tEnv,"Dwd_03_TradeOrderDetail");
        Table sourceInfo = tEnv.sqlQuery("select " +
                "   `common`['sid'] session_id," +
                "   `common`['sc'] source_id " +
                "from ods_log");

//        sourceInfo.execute().print();
        tEnv.createTemporaryView("source_info",sourceInfo);
    }

    public void readOdsLog(StreamTableEnvironment tEnv, String groupId) {
        tEnv.executeSql("create table ods_log (" +
                "   `common` map<string,string>," +
                "   `ts` bigint," +
                "   `pt` as proctime()" +
                ")" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_ODS_LOG, groupId));

    }

}
